<?php

/**
 * Allows accessing a traversable variable as a stream using callbacks
 *
 * PHP version 5
 *
 * LICENSE:
 *
 * Copyright (c) 2008, Philippe Jausions / 11abacus
 *
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   - Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   - Redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   - Neither the name of 11abacus nor the names of its contributors may
 *     be used to endorse or promote products derived from this software
 *     without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * @category  Stream
 * @package   Stream_Iterate
 * @author    Philippe Jausions <Philippe.Jausions@11abacus.com>
 * @copyright 2008 by Philippe Jausions / 11abacus
 * @license   http://www.opensource.org/licenses/bsd-license.php New BSD
 * @link      http://pear.11abacus.com/package/Stream_Iterate
 */

/**
 * Allows accessing a traversable variable as a stream using callbacks
 *
 * To access an iterator as a stream wrapper you'll need:
 * <ul>
 *  <li>The Stream_Iterate class</li>
 *  <li>Register the stream with the Stream_Iterate class</li>
 *  <li>Create a context containing the iterator 'subject'</li>
 *  <li>A callback function/method to convert 'subject' of the
 *      iteration into a string</li>
 * </ul>
 *
 * Example:
 * <code>
 *  // We need the Stream_Iterate class
 *  require_once 'Stream/Iterate.php';
 *
 *  // Register class as "foreach"
 *  if (!stream_wrapper_register('foreach', Stream_Iterate)) {
 *      echo 'Try a different protocol name, or use Stream_Iterate::register() without argument.';
 *      exit;
 *  }
 *
 *  // Some data we can iterate over with foreach (array, Traversable, etc...)
 *  // We'll assume MDB2 connection is configured somewhere else
 *  $db = MDB2::singleton();
 *  if (PEAR::isError($db)) {
 *      echo $db->getMessage();
 *      exit;
 *  }
 *  $result = $db->query('SELECT * FROM sometable');
 *  if (PEAR::isError($result)) {
 *      echo $result->getMessage();
 *      exit;
 *  }
 *  $data = new MDB2_Iterate($result);
 *
 *  // Let's define a callback to convert the data into a string that will be
 *  // piped into the stream (any valid callback will do)
 *  $callback = create_function('$key, $value',
 *                              'return $key."=".implode(',',$value)."\r\n";');
 *
 *  // Create the context for the stream
 *  $context = stream_context_create(array('foreach' => array(
 *                                                'subject'  => &$data,
 *                                                'callback' => $callback)));
 *  // Open the stream
 *  $fp = fopen('foreach://subject', 'r', 0, $context);
 *  if (!$fp) {
 *      echo 'Could not open path';
 *      exit;
 *  }
 *
 *  // Let's use HTTP_Download to prompt user to save data
 *  HTTP_Download::staticSend(array('resource' => $fp, 'gzip' => true));
 * </code>
 *
 * URL examples (assuming "foreach" is the name of the registered protocol):
 * <ul>
 *  <li>foreach://key:value@subject</li>
 *  <li>foreach://value:key@subject</li>
 *  <li>foreach://key@subject</li>
 *  <li>foreach://value@subject</li>
 *  <li>foreach://subject</li>
 * </ul>
 * If not specified, "key:value" will be assumed. So protocol://subject
 * is equivalent to protocol://key:value@subject
 *
 * To each of the URL examples above you can append a query string made up of:
 * <ul>
 *  <li>toString=function_name</li>
 *  <li>toString=class::method</li>
 *  <li>formatter=class_name (an instance of the that class will be created
 *      and used as formatter; see Stream_Iterate_FormatterInterface)</li>
 * </ul>
 *
 * @category  Stream
 * @package   Stream_Iterate
 * @author    Philippe Jausions <Philippe.Jausions@11abacus.com>
 * @copyright 2008 by Philippe Jausions / 11abacus
 * @license   http://www.opensource.org/licenses/bsd-license.php New BSD
 * @version   Release: @package_version@
 * @link      http://pear.11abacus.com/package/Stream_Iterate
 */
class Stream_Iterate
{
    /**
     * Context
     *
     * @var resource
     */
    public $context;

    /**
     * Iterator
     *
     * @var Traversable
     */
    protected $iterator;

    /**
     * Content buffer
     *
     * @var string
     */
    protected $buffer = '';

    /**
     * Position in the stream
     *
     * @var integer
     */
    protected $pos = 0;
    
    /**
     * Iteration counter
     *
     * @var integer
     */
    protected $iteration = 0;

    /**
     * Error reporting mode
     *
     * @var boolean
     */
    protected $reportErrors;

    /**
     * Callable formatter
     *
     * @var string
     */
    protected $toString = null;

    /**
     * Callback to call when first data is read
     *
     * @var callback
     */
    protected $onFirst = null;
    
    /**
     * Callback to call after last data is read
     *
     * @var callback
     */
    protected $onLast = null;
    
    /**
     * Parameter list
     *
     * @var array
     */
    protected $params = array('key', 'value');

    /**
     * Opens the stream
     *
     * This method is called by fopen() through the stream API
     *
     * @param string  $path         "URL" to open
     * @param string  $mode         "r" or "rb"
     * @param integer $flags        Pass STREAM_REPORT_ERRORS to have the
     *                              package call trigger_error() if an error
     *                              occurs.
     * @param string  &$opened_path Not used in this class
     *
     * @return boolean TRUE on success, FALSE otherwise
     */
    public function stream_open($path, $mode, $flags, &$opened_path)
    {
        $this->reportErrors = ($flags & STREAM_REPORT_ERRORS);

        if (is_null($this->context)) {
            return $this->error('Missing stream context');
        }
        switch ($mode) {
        case 'r':
        case 'rb':
            break;
        default:
            return $this->error('Unsupported '.$mode.' mode');
        }

        $info = parse_url($path);
        if ($info === false) {
            return $this->error('Invalid path');
        }
        if (empty($info['host'])) {
            return $this->error('The name of the stream context option containing the data is missing in the path as the host');
        }

        $var      = $info['host'];
        $protocol = $info['scheme'];

        $params = array();
        if (isset($info['user'])) {
            $params[] = $info['user'];
        }
        if (isset($info['pass'])) {
            $params[] = $info['pass'];
        }
        foreach ($params as $p) {
            switch ($p) {
            case 'key':
            case 'value':
                break;
            default:
                return $this->error('Invalid user and/or password in path. Only "key" and "value" accepted.');
            }
        }
        $this->params = (!$params) ? array('key', 'value') : $params;

        $options = stream_context_get_options($this->context);

        if (empty($options[$protocol])) {
            return $this->error('No stream context options found for protocol');
        }

        $options = $options[$protocol];

        if (!array_key_exists($var, $options)) {
            return $this->error('The host in path does not match any option name in the stream context');
        }
        if (is_array($options[$var])) {
            $array = new ArrayObject($options[$var]);

            $this->iterator = $array->getIterator();
        } elseif ($options[$var] instanceof Iterator) {
            $this->iterator = $options[$var];
        } elseif ($options[$var] instanceof IteratorAggregate) {
            $this->iterator = $options[$var]->getIterator();
        } else {
            return $this->error('Data in stream context cannot be iterated over');
        }

        $qs = array();
        if (!empty($info['query'])) {
            $query = explode('&', $info['query']);
            foreach ($query as $q) {
                $data = explode('=', $q, 2);

                $qs[$data[0]] = urldecode((count($data) != 2) ? $data[0]
                                                              : $data[1]);
            }
        }
        $opt = array_merge($options, $qs);

        // Formatter class used?
        if (isset($opt['formatter'])) {
            if (!is_object($opt['formatter'])) {
                $opt['formatter'] = new $opt['formatter'];
            }
            if (!($opt['formatter'] instanceof Stream_Iterate_FormatterInterface)) {
                return $this->error('"formatter" option is not a Stream_Iterate_FormatterInterface instance');
            }
            $this->params    = array('key', 'value');
            $opt['onFirst']  = array(&$opt['formatter'], 'onFirstIteration');
            $opt['toString'] = array(&$opt['formatter'], 'formatKeyPair');
            $opt['onLast']   = array(&$opt['formatter'], 'onLastIteration');
        }

        $this->toString = null;
        foreach ($opt as $key => $value) {
            switch ($key) {
            case 'toString':
            case 'onFirst':
            case 'onLast':
                if (is_string($value) && strpos($value, '::') !== false) {
                    $value = explode('::', $value);
                }
                if (!is_callable($value)) {
                    return $this->error('"'.$key.'" option is not callable');
                }
                break;
            default:
                continue 2;
            }
            $this->$key = $value;
        }

        $this->iterator->rewind();

        return true;
    }

    /**
     * Calls trigger_error() whenever needed to honor the
     * STREAM_REPORT_ERRORS option
     *
     * @param string $msg error message
     *
     * @return boolean Always returns FALSE
     */
    protected function error($msg)
    {
        if ($this->reportErrors) {
            trigger_error($msg, E_USER_ERROR);
        }
        return false;
    }

    /**
     * Closes the stream
     *
     * This method is called by fclose() through the stream API
     *
     * @return void
     */
    public function stream_close()
    {
        unset($this->iterator);
        $this->buffer = '';
    }

    /**
     * Reads up to $count bytes from stream
     *
     * This method is called by fread() and fgets() through the stream API
     *
     * @param integer $count Number of bytes to read
     *
     * @return void
     */
    public function stream_read($count)
    {
        if ($count < 0) {
            return $this->error('Cannot read negative number of bytes');
        }
        if ($count == 0) {
            return '';
        }

        // Fill the buffer
        while (strlen($this->buffer) < $count
               && $this->iterator->valid()) {

            $this->read();
            ++$this->iteration;
            $this->iterator->next();
        }

        $return = (string)substr($this->buffer, 0, $count);
        
        // End of data?
        if (strlen($return) < $count && $this->onLast !== null) {
            $this->buffer .= call_user_func($this->onLast, $this->iteration - 1);
            $this->onLast  = null;

            // Make sure we use the eof data 
            $return = (string)substr($this->buffer, 0, $count);
        }

        $this->pos   += strlen($return);
        $this->buffer = (string)substr($this->buffer, $count);

        return $return;
    }

    /**
     * Gets the next iteration into the buffer
     *
     * Takes the value of the current iteration, passes it through callback
     * and appends the resulting string to the buffer
     *
     * @return void
     */
    protected function read()
    {
        $data = $this->iterator->current();

        $first = '';
        if ($this->iteration == 0 && $this->onFirst !== null) {
            $first = call_user_func($this->onFirst, $data);
        }

        if ($this->toString === null) {
            $data = (string)$data;
        } else {
            $params = array();
            foreach ($this->params as $p) {
                switch ($p) {
                case 'key':
                    $params[] = $this->iterator->key();
                    break;
                case 'value':
                    $params[] = $data;
                    break;
                }
            }
            // Add iteration counter as last parameter
            $params[] = $this->iteration;
            
            $data = call_user_func_array($this->toString, $params);
        }
        $this->buffer .= $first.$data;
    }

    /**
     * Writes to stream (unsupported)
     *
     * This method is called by fwrite() through the stream API
     *
     * @param string $data Data to write
     *
     * @return integer the number of bytes written to the stream
     */
    public function stream_write($data)
    {
        return $this->error('This stream is not writeable');
    }

    /**
     * EOF
     *
     * This method is called by feof() through the stream API
     *
     * @return boolean
     */
    public function stream_eof()
    {
        return !$this->iterator->valid();
    }

    /**
     * Returns position in the stream
     *
     * This method is called by ftell() through the stream API
     *
     * @return integer
     */
    public function stream_tell()
    {
        return $this->pos;
    }

    /**
     * Moves forward.in stream
     *
     * This method is called by fseek() through the stream API.
     *
     * @param integer $offset The offset
     * @param integer $whence One of SEEK_SET, SEEK_CUR or SEEK_END
     *
     * @return boolean Returns TRUE if position was updated, FALSE otherwise
     */
    public function stream_seek($offset, $whence = SEEK_SET)
    {
        switch ($whence) {
        case SEEK_END:
            return $this->error('This stream is not seekable from its end');
        case SEEK_CUR:
            $pos = $this->pos + $offset;
            break;
        case SEEK_SET:
        default:
            $pos = $offset;
        }
        if ($pos < $this->pos) {
            return $this->error('Cannot rewind stream');
        }
        if ($pos > $this->pos) {
            $this->stream_read($this->pos - $pos);
        }
        return true;
    }

    /**
     * Returns information about the stream
     *
     * No data is returned. This method is called by fstat() through the
     * stream API
     *
     * @return array
     */
    public function stream_stat()
    {
        return array(
            0  => 0,
            1  => 0,
            2  => 0100444,
            3  => 0,
            4  => -1,
            5  => -1,
            6  => -1,
            7  => -1,
            8  => 0,
            9  => 0,
            10 => 0,
            11 => -1,
            12 => -1,
            'dev'     => 0,
            'ino'     => 0,
            'mode'    => 0100444,
            'nlink'   => 0,
            'uid'     => -1,
            'gid'     => -1,
            'rdev'    => -1,
            'size'    => -1,
            'atime'   => 0,
            'mtime'   => 0,
            'ctime'   => 0,
            'blksize' => -1,
            'blocks'  => -1);
    }

    /**
     * Returns information about the path
     *
     * @param string  $path  Path to return information for
     * @param integer $flags Flags set by the Stream API
     *
     * @return array
     */
    public function url_stat($path, $flags)
    {
        return self::stream_stat();
    }
}

?>